package ScalaSparkApps


import org.apache.spark.{SparkConf, SparkContext}

object SparkDemo {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("ScalaSparkDemo").setMaster("yarn-client")
    val sc = new SparkContext(conf)
    val data1 = Array(1, 2, 3, 4, 5, 6, 6)
    val data2 = Array(9, 8, 7, 6, 5, 4, 3)
    val rdd1 = sc.parallelize(data1)
    val rdd2 = sc.parallelize(data2)

    /*------------这里是对单个RDD的转化操作-------------*/
    val afterMap = rdd1.map(x => x + 1)
    println("afterMap---" + afterMap.collect().mkString(","))//afterMap---2,3,4,5,6,7,7


    val afterFlatMap = rdd1.map(x => x - 1)
    println("afterFlatMap---" + afterFlatMap.collect.mkString(",")) //afterFlatMap---0,1,2,3,4,5,5


    val afterFilter = rdd1.filter(x => x > 2)
    println("afterFilter---" + afterFilter.collect().mkString(",")) //afterFilter---3,4,5,6,6


    val afterDisTinct = rdd1.distinct
    println("afterDisTinct---" + afterDisTinct.collect().mkString(",")) //afterDisTinct---4,6,2,1,3,5


    /*------------这里是对两个RDD的转化操作-------------*/
    val afterUnion = rdd1.union(rdd2)
    println("afterUnion---" + afterUnion.collect.mkString(",")) //afterUnion---1,2,3,4,5,6,6,9,8,7,6,5,4,3


    val afterIntersection = rdd1.intersection(rdd2)
    println("afterIntersection---" + afterIntersection.collect.mkString(",")) //afterIntersection---4,6,3,5


    val afterSubtarct = rdd1.subtract(rdd2)
    println("afterSubtarct---" + afterSubtarct.collect.mkString(",")) //afterSubtarct---2,1


    /*------------这里是对RDD的行为(action)操作----------*/
    println("collect--" + rdd1.collect.mkString(",")) //collect--1,2,3,4,5,6,6

    println("count---" + rdd1.count.toString) //count---7

    println("countByValue---" +rdd1.countByValue.mkString(",")) //countByValue---5 -> 1,1 -> 1,6 -> 2,2 -> 1,3 -> 1,4 -> 1

    println("take---" +rdd1.take(1).mkString(",")) //take---1

    println("top---" +rdd1.top(2).mkString(",")) //top---6,6

    println("takeOrdered---" +rdd1.takeOrdered(4).mkString(",")) //takeOrdered---1,2,3,4

    println("takeSample---" +rdd1.takeSample(false, 1).mkString(",")) //takeSample---6

    println("reduce---" +rdd1.reduce((x, y) => x + y).toString) //reduce---27

    println("fold---" +rdd1.fold(1)((x, y) => x + y).toString) //fold---30

    sc.stop()
  }
}
